package com.aotain.project.apollo.bolt;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Tuple;

import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.mushroom.Slave;
import com.aotain.apollo.utils.ApolloProperties;
import com.aotain.storm.AbstractBolt;

/**
 * 基础统计  延时 半小时
 * @author Administrator
 * 
 * Modified by turk 2017-06-29 增加总异常数据的统计记录
 *
 */
public class BaseStatSumBoltDelay extends AbstractBolt{


	/**
	 * 
	 */
	private static final long serialVersionUID = -8171942294850707240L;
	HashMap<String,Long> DestIPOCTETS_30M = new HashMap<String,Long>();
	HashMap<String,Long> DestIPOCTETS_HOUR = new HashMap<String,Long>();
	HashMap<String,Long> DestIPOCTETS_DAY = new HashMap<String,Long>();
	
	HashMap<String,Long> DestIPPACKET_30M = new HashMap<String,Long>();
	HashMap<String,Long> DestIPPACKET_HOUR = new HashMap<String,Long>();
	HashMap<String,Long> DestIPPACKET_DAY = new HashMap<String,Long>();
	
	HashMap<String,Long> DestIPPV_30M = new HashMap<String,Long>();
	HashMap<String,Long> DestIPPV_HOUR = new HashMap<String,Long>();
	HashMap<String,Long> DestIPPV_DAY = new HashMap<String,Long>();
	
	@Override
	public void cleanup() {
		// TODO Auto-generated method stub
		if(!zooserver.isEmpty())
	  	{
		  	Slave s = new Slave(zooserver);
		  
		  	s.SendAlarmLog("ERROR", "BaseStatSumBoltDelay Shutdown");
	  	}
	}

	@Override
	public void execute(Tuple input) {
		// TODO Auto-generated method stub
				try {
					
					if(isTickTuple(input)) {//满足时间条件
						emitCountingData();  //计算，入库
						collector.ack(input);
					} else {//不满足时间条件
						countInLocal(input); //数据记录 
						collector.ack(input);
					}
				} catch (Exception e) {
					// TODO Auto-generated catch block
					Logger.getRootLogger().error(e);
					throw new FailedException("BaseStatSumBolt Exception",e);
				}	
		
	}
	
	private void countInLocal(Tuple tuple) {
		
		//"IDC","IP","PV"
		String DestIP = tuple.getStringByField("IP");
//		int destport = Integer.parseInt(tuple.getString(1));
		
		String idc = tuple.getStringByField("IDC");
		
		String type = tuple.getStringByField("TYPE");
		
		long nOctets = tuple.getLongByField("OCTETS");
		
		long nPacket = tuple.getLongByField("PACKET");
		
		long nPV = tuple.getLongByField("PV");
		
		

		Date date = new Date();
	
		SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
		SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");
		//SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH");
		SimpleDateFormat sdf_min  = new SimpleDateFormat("mm");
		String dateStr_min = sdf_min.format(date);
		String dateStr_h = sdf_h.format(date);
		String dateStr_d = sdf_d.format(date);
		
		
		//##### 30MIN
		int nMin30 = (Integer.parseInt(dateStr_min) / 30) * 30;
		String sMin30 = String.format("%02d", nMin30);   
		String dateStr30 = dateStr_h + sMin30;
		String rowkey30 = String.format("%s|%s|%s", DestIP, dateStr30, type);
		if(DestIPOCTETS_30M.containsKey(rowkey30))
		{
			long dipOCTETS = 0;
			dipOCTETS = DestIPOCTETS_30M.get(rowkey30) + nOctets;
			DestIPOCTETS_30M.put(rowkey30, dipOCTETS);
			
			long dipPACKET = 0;
			dipPACKET = DestIPPACKET_30M.get(rowkey30) + nPacket;
			DestIPPACKET_30M.put(rowkey30, dipPACKET);
			
			long dipPV = 0;
			dipPV = DestIPPV_30M.get(rowkey30) + nPV;
			DestIPPV_30M.put(rowkey30, dipPV);
		}
		else
		{
			DestIPOCTETS_30M.put(rowkey30, nOctets);
			DestIPPACKET_30M.put(rowkey30, nPacket);
			DestIPPV_30M.put(rowkey30, nPV);
		}
		
		//##### 小时
		String rowkeyHour = String.format("%s|%s00|%s", DestIP, dateStr_h, type);
		if(DestIPOCTETS_HOUR.containsKey(rowkeyHour))
		{
			long dipOCTETS = 0;
			dipOCTETS = DestIPOCTETS_HOUR.get(rowkeyHour) + nOctets;
			DestIPOCTETS_HOUR.put(rowkeyHour, dipOCTETS);
			
			long dipPACKET = 0;
			dipPACKET = DestIPPACKET_HOUR.get(rowkeyHour) + nPacket;
			DestIPPACKET_HOUR.put(rowkeyHour, dipPACKET);
			
			long dipPV = 0;
			dipPV = DestIPPV_HOUR.get(rowkeyHour) + nPV;
			DestIPPV_HOUR.put(rowkeyHour, dipPV);
		}
		else
		{
			DestIPOCTETS_HOUR.put(rowkeyHour, nOctets);
			DestIPPACKET_HOUR.put(rowkeyHour, nPacket);
			DestIPPV_HOUR.put(rowkeyHour, nPV);
		}
		
		//###天
		//if(type.equals("PV"))
		{
			String rowkeyDay = String.format("%s|%s|%s|%s", DestIP, dateStr_d , idc, type);
			if(DestIPOCTETS_DAY.containsKey(rowkeyDay))
			{
				long dipOCTETS = 0;
				dipOCTETS = DestIPOCTETS_DAY.get(rowkeyDay) + nOctets;
				DestIPOCTETS_DAY.put(rowkeyDay, dipOCTETS);
				
				long dipPACKET = 0;
				dipPACKET = DestIPPACKET_DAY.get(rowkeyDay) + nPacket;
				DestIPPACKET_DAY.put(rowkeyDay, dipPACKET);
				
				long dipPV = 0;
				dipPV = DestIPPV_DAY.get(rowkeyDay) + nPV;
				DestIPPV_DAY.put(rowkeyDay, dipPV);
			}
			else
			{
				DestIPOCTETS_DAY.put(rowkeyDay, nOctets);
				DestIPPACKET_DAY.put(rowkeyDay, nPacket);
				DestIPPV_DAY.put(rowkeyDay, nPV);
			}
		}
		
	}
	
	
	private void emitCountingData() {
		
		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(zooserver);
		
		for(String key : DestIPOCTETS_30M.keySet()) {
			
			String destIP = key.split("\\|",-1)[0];
			
			String dateStr30 = key.split("\\|",-1)[1];
			String type = key.split("\\|",-1)[2];
			String rowKey = String.format("%s_%s", destIP,dateStr30);
			long nOct = DestIPOCTETS_30M.get(key);
				
			String tbName30 = ApolloProperties.SDS_SESSION_STAT_30MIN;
			hbaseInstance.Add(tbName30, rowKey, "cf",  "DESTIP", destIP);
			hbaseInstance.Add(tbName30, rowKey, "cf",  "REPORTTIME", dateStr30);
			hbaseInstance.incerment(tbName30, rowKey, "cf:" + type, nOct);
			if(!type.equals("PV") && !type.equals("SPV")) {
				hbaseInstance.incerment(tbName30, rowKey, "cf:ATTOCT", nOct);
			}
				
			long nPacket = DestIPPACKET_30M.get(key);
			if(nPacket != 0)
			{
				hbaseInstance.Add(tbName30, rowKey, "cf",  "DESTIP", destIP);
				hbaseInstance.Add(tbName30, rowKey, "cf",  "REPORTTIME", dateStr30);
				hbaseInstance.incerment(tbName30, rowKey, "cf:" + type + "NUM", nPacket);
				if(!type.equals("PV") && !type.equals("SPV")) {
					hbaseInstance.incerment(tbName30, rowKey, "cf:ATTPACKET", nPacket);
				}
			}
			
			long nPV = DestIPPV_30M.get(key);
			if(nPV != 0)
			{
				hbaseInstance.Add(tbName30, rowKey, "cf",  "DESTIP", destIP);
				hbaseInstance.Add(tbName30, rowKey, "cf",  "REPORTTIME", dateStr30);
				hbaseInstance.incerment(tbName30, rowKey, "cf:ATTACKNUM", nPV);
			}
		}
			
		for(String key : DestIPOCTETS_HOUR.keySet()) {
			
			String destIP = key.split("\\|",-1)[0];
			//if(_ImportIPSet!=null && _ImportIPSet.size() > 0 && _ImportIPSet.contains(destIP))
			{
				String dateStr_h = key.split("\\|",-1)[1];
				String type = key.split("\\|",-1)[2];
				long nOct = DestIPOCTETS_HOUR.get(key);
				
				String rowKey = String.format("%s_%s", destIP,dateStr_h);
				
				String tbNameHour = ApolloProperties.SDS_SESSION_STAT_HOUR;
				hbaseInstance.Add(tbNameHour, rowKey, "cf",  "DESTIP", destIP);
				hbaseInstance.Add(tbNameHour, rowKey, "cf",  "REPORTTIME", dateStr_h);
				hbaseInstance.incerment(tbNameHour, rowKey, "cf:" + type, nOct);
				if(!type.equals("PV") && !type.equals("SPV")) {
					hbaseInstance.incerment(tbNameHour, rowKey, "cf:ATTOCT", nOct);
				}
				
				long nPacket = DestIPPACKET_HOUR.get(key);
				if(nPacket != 0)
				{
					hbaseInstance.Add(tbNameHour, rowKey, "cf",  "DESTIP", destIP);
					hbaseInstance.Add(tbNameHour, rowKey, "cf",  "REPORTTIME", dateStr_h);
					hbaseInstance.incerment(tbNameHour, rowKey, "cf:" + type + "NUM", nPacket);
					if(!type.equals("PV") && !type.equals("SPV")) {
						hbaseInstance.incerment(tbNameHour, rowKey, "cf:ATTPACKET", nPacket);
					}
				}
				
				long nPV = DestIPPV_HOUR.get(key);
				if(nPV != 0)
				{
					hbaseInstance.Add(tbNameHour, rowKey, "cf",  "DESTIP", destIP);
					hbaseInstance.Add(tbNameHour, rowKey, "cf",  "REPORTTIME", dateStr_h);
					hbaseInstance.incerment(tbNameHour, rowKey, "cf:ATTACKNUM", nPV);
				}
			}
		}
		
		for(String key : DestIPOCTETS_DAY.keySet()) {
			
			//######  DAY  #####
				String destIP = key.split("\\|",-1)[0];
			
				String dateStr_d = key.split("\\|",-1)[1];
				String IDC = key.split("\\|",-1)[2];
				String type = key.split("\\|",-1)[3];
				
				long nOct = DestIPOCTETS_DAY.get(key);
				
				//###2016-03-11 IDC 管理需求 
				String rowkeyIDCDay = String.format("%s_%s_%s", dateStr_d, IDC, destIP);
				String tbNameIDCDay = ApolloProperties.SDS_SESSION_STAT_DAY;
				hbaseInstance.Add(tbNameIDCDay, rowkeyIDCDay, "cf",  "DESTIP", destIP);
				hbaseInstance.Add(tbNameIDCDay, rowkeyIDCDay, "cf",  "IDC", IDC);
				hbaseInstance.Add(tbNameIDCDay, rowkeyIDCDay, "cf",  "REPORTTIME", dateStr_d);
				hbaseInstance.incerment(tbNameIDCDay, rowkeyIDCDay, "cf:" + type, nOct);
				if(!type.equals("PV") && !type.equals("SPV")) {
					hbaseInstance.incerment(tbNameIDCDay, rowkeyIDCDay, "cf:ATTOCT", nOct);
				}
				
				//###2016-08-22 公有云方案 add
				String rowkeyIPDay = String.format("%s_%s", destIP, dateStr_d);
//				String tbNameIDCDay = ApolloProperties.SDS_SESSION_STAT_DAY;
				hbaseInstance.Add(tbNameIDCDay, rowkeyIPDay, "cf",  "DESTIP", destIP);
				hbaseInstance.Add(tbNameIDCDay, rowkeyIPDay, "cf",  "IDC", IDC);
				hbaseInstance.Add(tbNameIDCDay, rowkeyIPDay, "cf",  "REPORTTIME", dateStr_d);
				hbaseInstance.incerment(tbNameIDCDay, rowkeyIPDay, "cf:" + type, nOct);
				if(!type.equals("PV") && !type.equals("SPV")) {
					hbaseInstance.incerment(tbNameIDCDay, rowkeyIDCDay, "cf:ATTOCT", nOct);
				}
				
				
				long nPacket = DestIPPACKET_DAY.get(key);
				if(nPacket != 0)
				{
					hbaseInstance.Add(tbNameIDCDay, rowkeyIPDay, "cf",  "DESTIP", destIP);
					hbaseInstance.Add(tbNameIDCDay, rowkeyIPDay, "cf",  "REPORTTIME", dateStr_d);
					hbaseInstance.incerment(tbNameIDCDay, rowkeyIPDay, "cf:" + type + "NUM", nPacket);
					if(!type.equals("PV") && !type.equals("SPV")) {
						hbaseInstance.incerment(tbNameIDCDay, rowkeyIPDay, "cf:ATTPACKET", nPacket);
					}
				}
				
				long nPV = DestIPPV_DAY.get(key);
				if(nPV != 0)
				{
					hbaseInstance.Add(tbNameIDCDay, rowkeyIPDay, "cf",  "DESTIP", destIP);
					hbaseInstance.Add(tbNameIDCDay, rowkeyIPDay, "cf",  "REPORTTIME", dateStr_d);
					hbaseInstance.incerment(tbNameIDCDay, rowkeyIPDay, "cf:ATTACKNUM", nPV);
				}
		}
		
		
		DestIPOCTETS_HOUR.clear();
		DestIPOCTETS_30M.clear();
		DestIPOCTETS_DAY.clear();
		
		DestIPPACKET_30M.clear();
		DestIPPACKET_HOUR.clear();
		DestIPPACKET_DAY.clear();
		
		DestIPPV_30M.clear();
		DestIPPV_HOUR.clear();
		DestIPPV_DAY.clear();
		
	}


	@Override
	public void declareOutputFields(OutputFieldsDeclarer arg0) {
		// TODO Auto-generated method stub
		
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,600);
		return conf;
	}
	

	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		
	}

}
